import json
import time
from datetime import datetime

from kafka import KafkaConsumer, KafkaProducer


class KProducer:
    def __init__(self, bootstrap_servers, topic):
        """
        kafka 生产者
        :param bootstrap_servers: 地址
        :param topic:  topic
        """
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda m: json.dumps(m).encode('ascii'), )  # json 格式化发送的内容
        self.topic = topic

    def sync_one_producer(self, data):
        """
        同步发送 数据
        :param data_li:  发送数据
        :return:
        """
        future = self.producer.send(self.topic, data)
        # record_metadata = future.get(timeout=10)  # 同步确认消费
        # partition = record_metadata.partition  # 数据所在的分区
        # offset = record_metadata.offset  # 数据所在分区的位置
        # print('save success, partition: {}, offset: {}'.format(partition, offset))

    def sync_producer(self, data_li: list):
        """
        同步发送 数据
        :param data_li:  发送数据
        :return:
        """
        for data in data_li:
            future = self.producer.send(self.topic, data)
            record_metadata = future.get(timeout=10)  # 同步确认消费
            partition = record_metadata.partition  # 数据所在的分区
            offset = record_metadata.offset  # 数据所在分区的位置
            # print('save success, partition: {}, offset: {}'.format(partition, offset))

    def asyn_producer(self, data_li: list):
        """
        异步发送数据
        :param data_li:发送数据
        :return:
        """
        for data in data_li:
            self.producer.send(self.topic, data)
        self.producer.flush()  # 批量提交

    def asyn_producer_callback(self, data_li: list):
        """
        异步发送数据 + 发送状态处理
        :param data_li:发送数据
        :return:
        """
        for data in data_li:
            self.producer.send(self.topic, data).add_callback(self.send_success).add_errback(self.send_error)
        self.producer.flush()  # 批量提交

    def send_success(self, *args, **kwargs):
        """异步发送成功回调函数"""
        print('save success')
        return

    def send_error(self, *args, **kwargs):
        """异步发送错误回调函数"""
        print('save error')
        return

    def close_producer(self):
        try:
            self.producer.close()
        except:
            pass


if __name__ == '__main__':

    kp = KProducer(topic='123_124_back_traffic_flow', bootstrap_servers='10.102.1.113:9092')
    while True:
        # 打开文件
        with open(r'E:\01文档\00仿真项目\数据集\webgl_simulation\test\111.txt', 'r') as file:
            # 逐行读取文件内容
            for line in file:
                # json_data = json.loads(line)
                # timeStamp = json_data['timeStamp']
                # date_obj = datetime.strptime(timeStamp, "%Y.%m.%d-%H.%M.%S.%f")
                # # 获取时间戳（秒级），并转换为毫秒级时间戳
                # timestamp_ms = int(date_obj.timestamp() * 1000)
                # 打印每一行内容
                # print(line.strip())  # strip()方法用于删除行末尾的换行符
                time.sleep(0.1)
                print('背景交通流发送中。。。。')
                # 同步发送
                kp.sync_one_producer(line.strip())
    # kp.close_producer()
